package com.tjbklx1.redis.chap06;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.UUID;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Tuple;

import com.google.gson.Gson;
import com.tjbklx1.redis.JedisUtil;

public class App65_DelayedTask {

	public static void main(String[] args) throws Exception {
		new App65_DelayedTask().run();
	}

	private void run() throws Exception {
		Jedis conn = JedisUtil.getJedisWithDB();
		testDelayedTasks(conn);
	}

	public void testDelayedTasks(Jedis conn) throws InterruptedException {
		System.out.println("\n----- testDelayedTasks -----");
		conn.del("queue:tqueue", "delayed:");
		System.out.println("Let's start some regular and delayed tasks...");
		for (long delay : new long[] { 0, 500, 0, 1500 }) {
			assert executeLater(conn, "tqueue", "testfn", new ArrayList<String>(), delay) != null;
		}
		long r = conn.llen("queue:tqueue");
		System.out.println("How many non-delayed tasks are there (should be 2)? " + r);
		assert r == 2;
		System.out.println();

		System.out.println("Let's start up a thread to bring those delayed tasks back...");
		PollQueueThread thread = new PollQueueThread();
		thread.start();
		System.out.println("Started.");
		System.out.println("Let's wait for those tasks to be prepared...");
		Thread.sleep(2000);
		thread.quit();
		thread.join();
		r = conn.llen("queue:tqueue");
		System.out.println("Waiting is over, how many tasks do we have (should be 4)? " + r);
		assert r == 4;
		conn.del("queue:tqueue", "delayed:");
	}

	public String executeLater(Jedis conn, String queue, String name, List<String> args, long delay) {
		Gson gson = new Gson();
		String identifier = UUID.randomUUID().toString();
		String itemArgs = gson.toJson(args);
		String item = gson.toJson(new String[] { identifier, queue, name, itemArgs });
		if (delay > 0) {
			conn.zadd("delayed:", System.currentTimeMillis() + delay, item); //延迟执行的任务
		} else {
			conn.rpush("queue:" + queue, item); //立即执行的任务
		}
		return identifier;
	}

	public class PollQueueThread extends Thread {
		private Jedis conn;
		private boolean quit;
		private Gson gson = new Gson();

		public PollQueueThread() {
			this.conn = new Jedis("localhost");
			this.conn.select(15);
		}

		public void quit() {
			quit = true;
		}

		public void run() {
			while (!quit) {
				Set<Tuple> items = conn.zrangeWithScores("delayed:", 0, 0); // 从延迟队列中取出一个对象
				Tuple item = items.size() > 0 ? items.iterator().next() : null;
				if (item == null || item.getScore() > System.currentTimeMillis()) {
					try {
						sleep(10);
					} catch (InterruptedException ie) {
						Thread.interrupted();
					}
					continue;
				}

				String json = item.getElement();
				String[] values = gson.fromJson(json, String[].class);
				String identifier = values[0];
				String queue = values[1];

				String locked = App63_DistributedLock.acquireLock(conn, identifier); //加锁
				if (locked == null) {
					continue;
				}

				if (conn.zrem("delayed:", json) == 1) { //从延迟队列删除
					conn.rpush("queue:" + queue, json); //加入立即执行的队列
				}

				App63_DistributedLock.releaseLock(conn, identifier, locked); // 释放锁
			}
		}
	}
}
